import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Random;
import java.util.Vector;


public class Client extends Machine implements IClient {

	IMachine master = null;
	ConnectionManager masterConnection = null,slaveConnection = null;
	Vector<KeySpaceObject> keySpace = null;
	MessageDigest md = null;
	static boolean first = true;
	
	public Client(String masterip) 
	{
		try {
			md = MessageDigest.getInstance("sha-1");
		} catch (NoSuchAlgorithmException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		keySpace = new Vector<KeySpaceObject>();
		master = new Machine();
		master.setIp(masterip);
		master.setId("MASTER");
		masterConnection = new ConnectionManager(PORTS.MASTER_LISTENER_PORT);
		masterConnection.connectTo(master);
		slaveConnection = new ConnectionManager(PORTS.SLAVE_LISTENER_PORT_FOR_CLIENT);
		//connManager.connectTo(master);		
	}
	
	public void populateKeySpace(byte[] data) throws IOException
	{
		this.keySpace.removeAllElements();
		ByteArrayInputStream bis = new ByteArrayInputStream(data);
		int total = bis.read();
		byte[] intbytes = new byte[4];
		for(int i = 0;i < total;i++){			
			bis.read(intbytes);
			int point = Message.getIntForBytes(intbytes);
			int strlen = bis.read();
			byte[] arr = new byte[strlen];
			bis.read(arr, 0, strlen);
			String ip = new String(arr);
			
			int portlen = bis.read();
			byte[] arr2 = new byte[portlen];
			bis.read(arr2, 0, portlen);
			String port = new String(arr2);
			KeySpaceObject ob = new KeySpaceObject(point, ip,port);
			this.keySpace.add(ob);
		}
		printKeySpace();
	}
	
	public KeySpaceObject findSlave(int hashedKey)
	{
		if(keySpace.size() <= 0)
			 return null;
		/*if(first){
			first = false;
			return keySpace.get(0);
		}else{
			first = true;
			return keySpace.get(1);
		}*/
		
		  
		for(int i = 0;i < keySpace.size();i++){
			if(keySpace.get(i).getPoint() >= hashedKey)
				return keySpace.get(i);				
		}
		return keySpace.get(0);
	}
	
	public void printKeySpace()
	{
		for(int i = 0;i < keySpace.size();i++){
			System.out.println(keySpace.get(i));								
		}
		
	}
	
	@Override
	public DataObject getData(DataObject key) {
		int hashedKey = getHash(key);
		KeySpaceObject kob = findSlave(hashedKey);
		String slave = (kob == null ? null : kob.getSlaveip());
		if(slave == null){
			//need to contact the master and get the corresponding slave
			IMessage message = new Message();
			message.setSource(this);
			message.setDest(master);
			message.setMessageID(IMessage.CLIENT_INQUIRE);
			message.setKey(key);		
			message.setValue(new DummyDataObject());
			
			masterConnection.writeData(master, message.serialize());
			IMessage response = null;
			try {
				response = masterConnection.receiveMessage();
				//populate keyspace
				populateKeySpace(response.getValue().getVal());
			}
			catch (IOException e) {
				e.printStackTrace();
			}
			kob = findSlave(hashedKey);
			slave = kob.getSlaveip();			
			//((CacheManager)cache).putPort(sl.getSlaveip(), sl.getPort());
		}
		//we should have got the slave by now. now contact the slave directly
		IMessage ms = new Message();
		ms.setSource(this);		
		ms.setMessageID(IMessage.GETDATA);
		/*String valueString = new String(key.getVal()) + ":':" + new Integer(hashedKey).toString();
		ms.setKey(new DataObject(valueString.getBytes()));//send both key and hashed key for the slave to use*/
		((Message)ms).setHashedKey(hashedKey);
		//String valueString = new String(key.getVal()) + ":':" + new Integer(hashedKey).toString();
		//ms.setKey(new DataObject(valueString.getBytes()));//send both key and hashed key for the slave to use
		ms.setKey(key);
		ms.setValue(new DummyDataObject());
		IMachine dest = new Machine();
		dest.setIp(slave);
		ms.setDest(dest);
		slaveConnection.connectTo(dest,new Integer(kob.getPort()).intValue());
		slaveConnection.writeData(dest, ms.serialize());
		
		//handle the response from slave
		IMessage dataResponse = null;
		try {
			dataResponse = slaveConnection.receiveMessage();
		} 
		catch (IOException e) {
			e.printStackTrace();
		}
		DataObject retval = dataResponse.getValue();
		terminateSlaveConnection();
		return retval;
	}
	
	public void terminateSlaveConnection()
	{
		IMessage msg = new Message();
		msg.setMessageID(IMessage.CLIENT_TERMINATE);
		msg.setDest(null);
		msg.setSource(this);
		msg.setKey(new DummyDataObject());
		msg.setValue(new DummyDataObject());
		slaveConnection.writeData(master, msg.serialize());		
		//slaveConnection.disconnectFrom(null);
	}

	@Override
	public int getHash(DataObject key) {
		int retVal = 0;
		// TODO Auto-generated method stub
		/*md.reset();
		byte[] digest = md.digest(key.getVal());
		DataInputStream is = new DataInputStream(new ByteArrayInputStream(digest));
		try {
			int a = is.readUnsignedByte();
			int b = is.readUnsignedByte();
			int c = is.readUnsignedByte();
			int d = is.readUnsignedByte();
			retVal = (((a & 0x3f) << 24) | ((b & 0xff) << 16) |
					  ((c & 0xff) << 8) | (d & 0xff));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
		retVal = Message.getIntForBytes(key.getVal());
		return retVal;
	}

	@Override
	public void putData(DataObject key, DataObject value) {
		int hashedKey = getHash(key);
		//System.out.println("hashkey : " + hashedKey);
		KeySpaceObject kob = findSlave(hashedKey);
		String slave = (kob == null ? null : kob.getSlaveip());
		if(slave == null){
			//need to contact the master and get the corresponding slave
			IMessage message = new Message();
			message.setSource(this);
			message.setDest(master);
			message.setMessageID(IMessage.CLIENT_INQUIRE);
			message.setKey(new DummyDataObject());
			message.setValue(new DummyDataObject());
			masterConnection.writeData(master, message.serialize());
			IMessage response = null;
			try {
				response = masterConnection.receiveMessage();
				populateKeySpace(response.getValue().getVal());
			}
			catch (IOException e) {
				e.printStackTrace();
			}
			kob = findSlave(hashedKey); 
			slave = kob.getSlaveip();			
		}
		//we should have got the slave by now. now contact the slave directly
		IMessage ms = new Message();
		ms.setSource(this);
		ms.setMessageID(IMessage.PUTDATA);
		((Message)ms).setHashedKey(hashedKey);		
		//String valueString = new String(key.getVal()) + ":':" + new Integer(hashedKey).toString();
		//ms.setKey(new DataObject(valueString.getBytes()));//send both key and hashed key for the slave to use
		ms.setKey(key);
		ms.setValue(value);
		IMachine dest = new Machine();
		dest.setIp(slave);
		ms.setDest(dest);
		slaveConnection.connectTo(dest,new Integer(kob.getPort()).intValue());
		//System.out.println("writing data " + ms);
		slaveConnection.writeData(dest, ms.serialize());
		//wait for the acknowledgement
		try {
			slaveConnection.receiveMessage();
		} catch (IOException e) {
			e.printStackTrace();
		}
		terminateSlaveConnection();		
	}
	
	public void closeAllConnections()
	{
		IMessage msg = new Message();
		msg.setMessageID(IMessage.CLIENT_TERMINATE);
		msg.setDest(master);
		msg.setSource(this);
		msg.setKey(new DummyDataObject());
		msg.setValue(new DummyDataObject());
		masterConnection.writeData(master, msg.serialize());
		masterConnection.disconnectFrom(master);
		slaveConnection.disconnectFrom(null);
	}
	
	public void generateKeysValues(byte[][] keys,byte[][] values,int n,int min,int max)
	{
		//use the same values for all the keys
		int keylen = 4,vallen = 1000*1024;//r.nextInt(100);
		Random r = new Random(System.currentTimeMillis());
		values[0] = new byte[vallen];
		Random r2 = new Random(r.nextLong());
		r2.nextBytes(values[0]);
		for(int i = 0;i < n;i++){		
			keys[i] = new byte[keylen];				
			keys[i] = Message.getBytesForInt(min + r.nextInt(max-min+1));
		}
	}
	
	public void printByteArray(byte[] x)
	{
		for(int i = 0;i < x.length;i++)
			System.out.print(x[i] + ",");
		System.out.println();
	}
	
	public static void main(String[] args) throws IOException//first argument should be the ip of the client, second argument should be the number of keys, thrid arg is the master ip ,fourth arg is the min value of key,fifth value is the max value of key,sixth arg is the logging file name
	{
		Client c = new Client(args[2]);
		c.setIp(args[0]);
		c.setId(args[0]);
		int i = 0,n = Integer.parseInt(args[1]);
		byte[][] keys = new byte[n][];
		byte[][] values = new byte[1][];
		c.generateKeysValues(keys,values,n,Integer.parseInt(args[3]),Integer.parseInt(args[4]));
		FileWriter logfile = new FileWriter(args[5]);
		logfile.write(new Long(System.currentTimeMillis()).toString());
		while(i < n){
			c.putData(new DataObject(keys[i]),new DataObject(values[0]));
			if(i % 100 == 0)
				System.out.println("100 keys done");
			//String val = new String((c.getData(new DataObject(new String("abcd").getBytes()))).getVal());
			//System.out.println("value got for key hello is " + val);
			//System.in.read();
			i++;
		}
		logfile.write(new Long(System.currentTimeMillis()).toString());
		DataObject readFromDB = null;
		int error = 0;
		for(i = 0;i < n;i++){
			readFromDB = c.getData(new DataObject(keys[i]));
			if(!readFromDB.isEqual(values[0])){
				System.out.println("error for item " + i);
				/*System.out.println("value written to DB: ");
				c.printByteArray(values[i]);*/
				System.out.println("from DB:");
				c.printByteArray(readFromDB.getVal());
				error++;
			}
		}
		logfile.write(new Long(System.currentTimeMillis()).toString());
		System.out.println("error count is " + error);
		logfile.close();
		c.closeAllConnections();
	}

}
